package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.article.msg.ArticleBehaviorMsg;
import com.heima.article.msg.ArticleBehaviorStreamMsg;
import com.heima.common.constants.message.HotArticleConstants;
import com.heima.common.util.JsonUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.SendTo;

import java.time.Duration;
import java.util.Date;

@Slf4j
@EnableBinding(BehaviorStreamProcessor.CountNumProcessor.class)
public class BehaviorStreamProcessor {

    @StreamListener(HotArticleConstants.HOT_ARTICLE_SCORE_STREAM_TOPIC)
    @SendTo(HotArticleConstants.HOT_ARTICLE_SCORE_CONSUMER_TOPIC)
    public KStream<String, String> process(KStream<String, String> input) {

        // 做了数据格式的转换，把key从value中拆分出来，方便下一步做分组聚合
        return input.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
                    @Override
                    public KeyValue<String, String> apply(String key, String value) {
                        log.info("value = {}", value);
//                        if (value.length() < 10){
//                            ArticleBehaviorMsg msg = new ArticleBehaviorMsg();
//                            msg.setArticleId(-1L);
//                            msg.setType(ArticleBehaviorTypeEnum.VIEWS);
//                            return new KeyValue<String, String>(msg.getArticleId() + "", JSON.toJSONString(msg));
//                        }
                        ArticleBehaviorMsg msg = JsonUtils.toBean(value, ArticleBehaviorMsg.class);
                        return new KeyValue<String, String>(msg.getArticleId() + "", value);
                    }
                    // 做分组聚合
                }).groupByKey(Grouped.with(Serdes.String(), Serdes.String()))
                // 设置时间窗口, 十秒为单位
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)).grace(Duration.ZERO))
                // 设置聚合算法
                .aggregate(new Initializer<String>() {
                    @Override
                    public String apply() {
                        log.info("--时间窗口初始化");
                        return null;
                    }
                }, new Aggregator<String, String, String>() {
                    // 时间窗口内的聚合
                    @Override
                    public String apply(String key, String value, String aggregate) {
                        log.info(new Date() + "时间窗口内的聚合, 当前key = " + key
                                + "当前value = " + value + "当前aggregate = " + aggregate);
                        // 解析出msg对象
                        ArticleBehaviorMsg msg = JsonUtils.toBean(value, ArticleBehaviorMsg.class);

                        ArticleBehaviorStreamMsg streamMsg;
                        if (aggregate == null) {
                            streamMsg = new ArticleBehaviorStreamMsg();
                            streamMsg.setArticleId(msg.getArticleId());
                        } else {
                            streamMsg = JsonUtils.toBean(aggregate, ArticleBehaviorStreamMsg.class);
                        }

                        switch (msg.getType()) {
                            case LIKES:
                                streamMsg.setLikes(streamMsg.getLikes() + msg.getNum());
                                break;
                            case VIEWS:
                                streamMsg.setViews(streamMsg.getViews() + msg.getNum());
                                break;
                            case COMMENT:
                                streamMsg.setComment(streamMsg.getComment() + msg.getNum());
                                break;
                            case COLLECTION:
                                streamMsg.setCollection(streamMsg.getCollection() + msg.getNum());
                                break;
                            default:
                                break;
                        }

                        return JSON.toJSONString(streamMsg);
                    }
                }, Materialized.with(Serdes.String(), Serdes.String()))
                // 只输出最终的计算结果, 不输出中间计算过程的数据
                .suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded()))
                // 重新修改结果的数据类型, 因为默认的key为时间窗口的类型
                .toStream().map(new KeyValueMapper<Windowed<String>, String, KeyValue<String, String>>() {
                    @Override
                    public KeyValue<String, String> apply(Windowed<String> stringWindowed, String value) {
                        String key = stringWindowed.key();
                        System.out.println("时间窗口聚合结束, key = " + key + "value = " + value);

                        return new KeyValue<>(key, value);
                    }
                });

    }

    public interface CountNumProcessor {
        @Input(HotArticleConstants.HOT_ARTICLE_SCORE_STREAM_TOPIC)
        KStream<String, String> input();

        @Output(HotArticleConstants.HOT_ARTICLE_SCORE_CONSUMER_TOPIC)
        KStream<String, String> output();
    }
}
